package com.wyj.rpc.config;

import com.wyj.rpc.entity.ZkDataBO;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.*;
import java.util.concurrent.*;

/**
 * TODO: rpc远程调用客户端，即netty的客户端。
 *  1、连接netty服务端
 *  2、发送请求
 *  3、关闭相关资源
 *
 * @author wuyijie
 * @date 2021/12/26 1:38 下午
 */
public class RpcClient {
    private static final String basePath = "/nettyServer";
    private static CuratorFramework zkClient = null;
    private static Map<InetSocketAddress, RpcClient> rpcClientMap = new HashMap<>();
    private static Map<InetSocketAddress, ZkDataBO> zkDataBOMap = new HashMap<>();


    NioEventLoopGroup nioEventLoopGroup;
    Channel channel;
    private InetSocketAddress inetSocketAddress;
    private RpcClientHandler rpcClientHandler;
    private ExecutorService executorService = Executors.newCachedThreadPool();

    /*
     * todo：从zk获取netty服务端信息，根据获取到的信息封装rpcClient
     * */
    static {
        //创建并启动zk客户端
        zkClient = CuratorFrameworkFactory.builder()
                .connectString("server.aliyun:12181")
                .connectionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 2000, 5))
                .build();
        zkClient.start();
        try {
            List<String> strings = zkClient.getChildren().forPath(basePath);
            //初始化
            //获取zk中的netty服务端信息并创建rpcClient
            //string:zknode名称
            for (String string : strings) {
                if (null != string && string.length() > 0) {
                    zkDataHandler(string);
                }
            }
            //添加zk监听
            final PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                    System.out.println("zk监听到的消息 pathChildrenCacheEvent=" + pathChildrenCacheEvent);
                    PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                    ChildData childData = pathChildrenCacheEvent.getData();
                    String childPath = childData.getPath().split(basePath + "/")[1];

                    InetSocketAddress inetSocketAddress = new InetSocketAddress(childPath.split(":")[0], Integer.parseInt(childPath.split(":")[1]));
                    //添加事件
                    if (type.equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
                        if (!rpcClientMap.containsKey(inetSocketAddress)) {
                            zkDataHandler(childPath);
                            System.out.println("zk服务端信息添加成功" + inetSocketAddress);
                        }
                    }
                    //删除事件
                    else if (type.equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                        rpcClientMap.remove(inetSocketAddress);
                        zkDataBOMap.remove(inetSocketAddress);
                        System.out.println("zk服务端信息删除成功" + inetSocketAddress);
                    }
                }
            };
            //设置监听
            final PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, basePath, true);
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);


            // final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
            // scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
            //     //zkNode：name=ip:port,data=totalTime_LocalDateTime.now()
            //     @Override
            //     public void run() {
            //         try {
            //             List<String> strings = zkClient.getChildren().forPath(basePath);
            //             for (String string : strings) {
            //                 if (null != string && string.length() > 0) {
            //                     String data = "_" + LocalDateTime.now().toString();
            //                     // updateNettyServer(string, data);
            //                 }
            //             }
            //         } catch (Exception e) {
            //             e.printStackTrace();
            //         }
            //     }
            // }, 2, 10, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private RpcClient(String ip, int port) {
        this.initClient(new InetSocketAddress(ip, port));
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        RpcClient rpcClient = (RpcClient) o;
        return this.inetSocketAddress == rpcClient.inetSocketAddress;
    }

    @Override
    public int hashCode() {
        return Objects.hash(inetSocketAddress);
    }

    /**
     * TODO: 获取zkNode数据并作处理
     *
     * @param zkNode zkNode 名称
     * @author wuyijie
     * @date 2022/1/7 9:45 PM
     */
    public static void zkDataHandler(String zkNode) {
        final byte[] bytes;
        try {
            InetSocketAddress inetSocketAddress = new InetSocketAddress(zkNode.split(":")[0], Integer.parseInt(zkNode.split(":")[1]));
            rpcClientMap.put(inetSocketAddress, new RpcClient(inetSocketAddress.getHostName(), inetSocketAddress.getPort()));
            bytes = zkClient.getData().forPath(basePath + "/" + zkNode);
            final var s = new String(bytes, StandardCharsets.UTF_8);
            if (s.length() > 0) {
                final var arr = s.split("_");
                zkDataBOMap.put(inetSocketAddress, new ZkDataBO(inetSocketAddress, "".equals(arr[0]) ? null : Long.valueOf(arr[0]), arr[1] == null ? LocalDateTime.now() : LocalDateTime.parse(arr[1])));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * TODO: 连接netty服务端
     *
     * @author wuyijie
     * @date 2021/12/26 1:47 下午
     */
    public void initClient(InetSocketAddress param) {
        rpcClientHandler = new RpcClientHandler();
        nioEventLoopGroup = new NioEventLoopGroup();
        this.inetSocketAddress = new InetSocketAddress(param.getHostName(), param.getPort());
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(nioEventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new StringDecoder());
                            //自定义客户端事件处理器
                            pipeline.addLast(rpcClientHandler);
                        }
                    });
            channel = bootstrap.connect(this.inetSocketAddress.getHostName(), this.inetSocketAddress.getPort()).sync().channel();
        } catch (InterruptedException e) {
            e.printStackTrace();
            if (null != channel) {
                channel.close();
            }
            if (null != nioEventLoopGroup) {
                nioEventLoopGroup.shutdownGracefully();
            }
        }
    }

    /**
     * TODO: 负载均衡---获取响应事件最小的服务端InetSocketAddress
     *
     * @return java.net.InetSocketAddress
     * @author wuyijie
     * @date 2022/1/7 9:36 PM
     */
    public static InetSocketAddress getSuitableInetSocketAddress() {
        System.out.println("zkDataBOMap=" + zkDataBOMap);
        final var entries = zkDataBOMap.entrySet();
        final var iterator = entries.iterator();
        Long totalTime = null;
        Map<Long, List<InetSocketAddress>> longListMap = new HashMap<>();
        while (iterator.hasNext()) {
            InetSocketAddress inetSocketAddress = null;
            final var next = iterator.next();
            if (totalTime == null) {
                totalTime = next.getValue().getTotalTime();
                inetSocketAddress = next.getValue().getInetSocketAddress();
            } else {
                if (null != next.getValue().getTotalTime()) {
                    if (totalTime > next.getValue().getTotalTime()) {
                        totalTime = next.getValue().getTotalTime();
                        inetSocketAddress = next.getValue().getInetSocketAddress();
                    }
                } else {
                    totalTime = next.getValue().getTotalTime();
                    inetSocketAddress = next.getValue().getInetSocketAddress();
                }
            }
            if (inetSocketAddress != null) {
                if (longListMap.containsKey(totalTime)) {
                    longListMap.get(totalTime).add(inetSocketAddress);
                } else {
                    List<InetSocketAddress> list = new ArrayList<>();
                    list.add(inetSocketAddress);
                    longListMap.put(totalTime, list);
                }
            }

            System.out.println("totalTime=" + totalTime);
        }
        final var random = new Random();
        if (longListMap.containsKey(null)) {
            totalTime = null;
        }
        final var inetSocketAddresses = longListMap.get(totalTime);
        final var i = random.nextInt(inetSocketAddresses.size());
        System.out.println("i=" + i);
        System.out.println("inetSocketAddress=" + inetSocketAddresses.get(i));
        return longListMap.get(totalTime).get(i);

    }


    /**
     * TODO: 同步数据到zk
     *
     * @param zkNodePath zkNodePath
     * @param data       data
     * @author wuyijie
     * @date 2022/1/7 9:58 PM
     */
    private static void updateNettyServer(String zkNodePath, String data) {
        final Stat stat;
        try {
            //如果节点不存在，创建临时节点
            zkNodePath = zkNodePath.replaceAll("localhost", "127.0.0.1");
            stat = zkClient.checkExists().forPath(basePath + "/" + zkNodePath);
            System.out.println("=====" + basePath + "/" + zkNodePath);
            System.out.println(stat);
            if (null != stat) {
                zkClient.setData().forPath(basePath + "/" + zkNodePath, data.getBytes(StandardCharsets.UTF_8));
                System.err.println("更新zkNode：" + zkNodePath + "，数据为：" + data);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * TODO: 发送请求
     *
     * @param msg rpc请求参数
     * @return java.lang.Object
     * @author wuyijie
     * @date 2021/12/26 2:00 下午
     */
    public Object sendRequest(String msg) throws ExecutionException, InterruptedException {
        final LocalDateTime begin = LocalDateTime.now();
        try {
            rpcClientHandler.setRequestMsg(msg);
            Future<Object> future = executorService.submit(rpcClientHandler);
            Object o = future.get();
            return o;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            LocalDateTime end = LocalDateTime.now();
            Duration between = Duration.between(begin, end);
            final var s = String.valueOf(between.toMillis() + "_" + LocalDateTime.now().toString());
            System.err.println("请求结束：" + s + "  " + inetSocketAddress);
            zkDataBOMap.put(inetSocketAddress, new ZkDataBO(inetSocketAddress, between.toMillis(), LocalDateTime.now()));
            updateNettyServer(inetSocketAddress.getHostName() + ":" + inetSocketAddress.getPort(), s);

        }
        return null;
    }

    public static RpcClient getInstance() {
        return rpcClientMap.get(getSuitableInetSocketAddress());
    }

    /**
     * TODO: 关闭资源
     *
     * @author wuyijie
     * @date 2021/12/26 2:10 下午
     */
    public void close() {
        if (null != channel) {
            channel.close();
        }
        if (null != nioEventLoopGroup) {
            nioEventLoopGroup.shutdownGracefully();
        }
    }
}
